package com.link.gateway.core.kafka;

import com.link.gateway.base.user.model.User;
import com.link.gateway.core.comproperty.RunProperties;
import com.link.gateway.core.kafka.conf.KafkaEnv;
import com.link.gateway.core.utils.CacheUtil;
import com.link.gateway.core.utils.SpringUtils;
import com.link.gateway.core.utils.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

public class KafkaTemplateFacotry {

    private static final Logger logger = LogManager.getLogger(KafkaTemplateFacotry.class);


    public static final String FILTER_LINKLOG_TOPIC = "KT_GATEWAY_FILTER_LOG";

    public static final String SEND_RESTRICTION_MAIL_TOPIC = "SEND_RESTRICTION_MAIL";

    public static final String LOG_TOPIC = "KT_LOG";

    /** Kafka开关*/
    public static final String KAFKA_ENABLED = "kafkaEnabled";



    private static KafkaTemplate<String, String> getInstance() {
        logger.info("获取kafkaTemplate");
        KafkaTemplate kafkaTemplate = SpringUtils.getBean(KafkaTemplate.class);
        return kafkaTemplate;
    }

    /**
     * 发送消息
     * @param topic
     * @param message
     */
    public static void send(String topic, String message) throws ExecutionException {
        //获取得到用户信息
        User user = RunProperties.TEMP_USER.get();
        String kafkaEnabled = null;
        boolean enabled = KafkaEnv.isEnabled();
        if (user != null) {
            //企业配置参数，Kafka开关
            Map<String, String> mapKey = new HashMap<>();
            mapKey.put("corpId", user.getCorpid().toString());
            mapKey.put("redisKey", KAFKA_ENABLED);
            kafkaEnabled = CacheUtil.getEnterpriseParamFromCache(mapKey);
        }
        if (StringUtils.isNotBlank(kafkaEnabled) && "N".equalsIgnoreCase(kafkaEnabled)) {
            enabled = false;
        }
        boolean saslEnabled = KafkaEnv.isSaslEnabled();
        if (enabled && saslEnabled) {
            try {
                KafkaTemplate<String, String> kafkaTemplate = getInstance();
                if (kafkaTemplate != null) {
                    kafkaTemplate.send(topic, message);
                }
            } catch (Exception e) {
                logger.error("发送消息失败！" + e.toString());
            }
        } else {
//            logger.warn("kafka server is not avilabel! enabled->" + enabled + ", saslEnabled->" + saslEnabled);
        }
    }
}
